package org.example.flinketl.util;

import java.lang.reflect.Method;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.commons.math3.geometry.euclidean.oned.Interval;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import ru.yandex.clickhouse.ClickHouseConnection;

public class ClickHouseInsertUtil<Object> extends RichSinkFunction<Object>{
  final int  batchSize = 5000;
  int unit = batchSize;
  long before = System.currentTimeMillis();
  long iniTime = System.currentTimeMillis();
  static ClickHouseConnection conn = null;
  long count = 0;
  static Map<Class<?>,PreparedStatement> preparedStatements = new HashMap<>();

  public ClickHouseInsertUtil() throws SQLException{
    if(preparedStatements == null)preparedStatements = new HashMap<>();
    conn = MyConfig.dataSource.getConnection();
  }


  @Override
  public void invoke(Object value, Context context) throws Exception {
    String sql = MyConfig.sqls.get(value.getClass());
    if(conn == null)conn = MyConfig.dataSource.getConnection();
    if(preparedStatements == null)System.out.println("prepareStatements is null");
    if(value == null)System.out.println("value is null");
    if(conn == null)System.out.println("conn is null");
    PreparedStatement preparedStatement = preparedStatements.getOrDefault(value.getClass(), conn.prepareStatement(sql));
    List<Method> methods = MyConfig.methods.get(value.getClass()); 
    try {
      // String sql = sqls.get(value.getClass().getSimpleName());
      int i = 1;
      for(Method m : methods){
        preparedStatement.setObject(i, m.invoke(value));
        i++;
      }
      preparedStatement.addBatch();
      preparedStatements.put(value.getClass(), preparedStatement);
      exectBatch();
      count++;
      unit--;
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  @Override
  public void close() throws Exception {
    // TODO Auto-generated method stub
    super.close();
  }

  @Override
  public void open(Configuration parameters) throws Exception {
    // TODO Auto-generated method stub
    super.open(parameters);
  }

  private void exectBatch(){
    if(unit>0)return;
    Set<Class<?>> temp = new HashSet<>(preparedStatements.keySet());
    for (Class<?> key : temp) {
      
      PreparedStatement preparedStatement = preparedStatements.remove(key);
      try {
        preparedStatement.executeBatch();
        conn.commit();
      } catch (SQLException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
      }
    }

    long now = System.currentTimeMillis();
    long interval = now - before;
    long totalInterval = now - iniTime;
    int rate = (int)(1.0 *  batchSize / interval * 1000);
    int rateTotal = (int)(1.0 * count / totalInterval * 1000);
    System.out.println("sink rate is " + rate + "/s" + " total is " + count + " average rate is " + rateTotal);
    before = now;
    unit = batchSize;
  }

}
